package com.winit.schedule.job.tracker;

import java.net.BindException;
import java.util.Properties;
import java.util.UUID;

import com.winit.schedule.job.tracker.sms.SmsSender;
import org.apache.commons.lang3.StringUtils;
import org.springframework.mail.javamail.JavaMailSenderImpl;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.event.AsyncDispatcher;
import com.winit.schedule.core.event.Dispatcher;
import com.winit.schedule.core.exception.UscheduleRuntimeException;
import com.winit.schedule.core.service.CompositeService;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.core.utils.NetUtils;
import com.winit.schedule.job.tracker.clean.DataCleanService;
import com.winit.schedule.job.tracker.conf.JobConfiguration;
import com.winit.schedule.job.tracker.event.ClusterEventType;
import com.winit.schedule.job.tracker.event.QuartzEventHandler;
import com.winit.schedule.job.tracker.event.QuartzEventType;
import com.winit.schedule.job.tracker.event.TaskEventHandler;
import com.winit.schedule.job.tracker.event.TaskEventType;
import com.winit.schedule.job.tracker.event.TaskTrackerEventHandler;
import com.winit.schedule.job.tracker.event.TaskTrackerEventType;
import com.winit.schedule.job.tracker.event.persist.DataPersistEventType;
import com.winit.schedule.job.tracker.event.persist.RunRecordHandler;
import com.winit.schedule.job.tracker.event.persist.TaskItemHandler;
import com.winit.schedule.job.tracker.event.persist.TaskTrackerHandler;
import com.winit.schedule.job.tracker.monitor.TaskTrackerMonitor;
import com.winit.schedule.job.tracker.monitor.ZKServiceMonitor;
import com.winit.schedule.job.tracker.monitor.ZkHealthCheckMonitor;
import com.winit.schedule.job.tracker.remoting.RemotingService;
import com.winit.schedule.job.tracker.repository.ConsoleDataManager;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;
import com.winit.schedule.job.tracker.schedule.EqualizerTaskManager;
import com.winit.schedule.job.tracker.schedule.TaskScheduleManager;
import com.winit.schedule.job.tracker.util.MailHelper;

public class JobTracker extends CompositeService {

    private JobTrackerContextImpl jobTrackerContext;

    private Dispatcher            dispatcher;

    private Dispatcher            dataDispatcher;

    private Configuration         conf;

    private RemotingService       remotingService;

    private ScheduleDataManager   scheduleDataManager;

    private ConsoleDataManager    consoleDataManager;

    private TaskTrackerMonitor    taskTrackerMonitor;

    private TaskScheduleManager   taskScheduleManager;

    private SmsSender             smsSender;

    private ZKServiceMonitor      zkService;

    private String                uuid;

    private String                localIP = "";

    public JobTracker(){
        super("JobTracker");
        localIP = NetUtils.getLocalIP();
        if (StringUtils.isBlank(localIP)) {
            throw new UscheduleRuntimeException(
                "Could not get the IP address of this machine,Please check the operating system hosts configuration file.");
        }
        this.uuid = localIP + "$" + (UUID.randomUUID().toString().replaceAll("-", "").toUpperCase());
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        this.conf = conf;

        int port = conf.getInt(JobConfiguration.LISTEN_PORT, JobConfiguration.DEFAULT_LISTEN_PORT);
        boolean used = NetUtils.isPortUsing(localIP, port);
        if (used) {
            throw new BindException(
                "Cannot assign requested address,the port already is used : uschedule_Bind:" + port);
        }

        this.jobTrackerContext = new JobTrackerContextImpl();
        jobTrackerContext.setJobTracker(this);
        jobTrackerContext.setConf(conf);

        // 构建事件分发器
        this.dispatcher = setupDispatcher();
        addIfService(dispatcher);
        jobTrackerContext.setDispatcher(dispatcher);

        this.dataDispatcher = setupDispatcher();
        addIfService(dataDispatcher);
        jobTrackerContext.setDataDispatcher(dataDispatcher);

        // 邮件发送
        MailHelper mailHelper = initMailHelper(conf);
        jobTrackerContext.setMailHelper(mailHelper);

        // 存储
        jobTrackerContext.setScheduleDataManager(this.scheduleDataManager);
        jobTrackerContext.setConsoleDataManager(this.consoleDataManager);
        jobTrackerContext.setSmsSender(this.smsSender);

        // 事件注册
        dispatcher.register(TaskEventType.class, new TaskEventHandler(this.jobTrackerContext));
        dispatcher.register(TaskTrackerEventType.class, new TaskTrackerEventHandler(this.jobTrackerContext));
        dispatcher.register(QuartzEventType.class, new QuartzEventHandler(this.jobTrackerContext));

        // 数据持久化分发
        dataDispatcher.register(DataPersistEventType.class, new RunRecordHandler(this.jobTrackerContext));
        dataDispatcher.register(DataPersistEventType.class, new TaskTrackerHandler(this.jobTrackerContext));
        dataDispatcher.register(DataPersistEventType.class, new TaskItemHandler(this.jobTrackerContext));

        // tasktracker 断开连接监听
        taskTrackerMonitor = new TaskTrackerMonitor(jobTrackerContext);
        addIfService(taskTrackerMonitor);
        jobTrackerContext.setTaskTrackerMonitor(taskTrackerMonitor);
        dispatcher.register(ClusterEventType.class, taskTrackerMonitor);

        // 任务调度管理
        taskScheduleManager = new EqualizerTaskManager(jobTrackerContext);
        jobTrackerContext.setTaskScheduleManager(taskScheduleManager);

        // 远程服务
        remotingService = new RemotingService(jobTrackerContext);
        addIfService(remotingService);
        jobTrackerContext.setRemotingService(remotingService);

        // 注册服务
        zkService = new ZKServiceMonitor(this.jobTrackerContext);
        addIfService(zkService);
        jobTrackerContext.setZkService(zkService);

        // 取消程序进行数据清理，改由DBA来处理
        // 数据清理
        //DataCleanService dataCleanService = new DataCleanService(jobTrackerContext);
        //addIfService(dataCleanService);

        // zookeeper注册地址定期检查，目前仅用于开发和测试环境
        boolean ckCheckEnable = conf.getBoolean(JobConfiguration.ZK_HEALTH_CHECK_ENABLE,
            JobConfiguration.DEFAULT_ZK_HEALTH_CHECK_ENABLE);
        boolean develoyMode = conf.getBoolean(JobConfiguration.DEVELOP_MODE,
                JobConfiguration.DEFAULT_DEVELOP_MODE);
        if (ckCheckEnable && !develoyMode) {
            ZkHealthCheckMonitor zkHealthCheckMonitor = new ZkHealthCheckMonitor(jobTrackerContext);
            addIfService(zkHealthCheckMonitor);
        }

        super.serviceInit(conf);
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        jobTrackerContext.setJobTrackerStartupTime(Clock.getSystemTime());
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
    }

    /**
     * Register the handlers for alwaysOn services
     */
    private Dispatcher setupDispatcher() {
        Dispatcher dispatcher = new AsyncDispatcher();
        return dispatcher;
    }

    /**
     * 初始化邮件服务
     * 
     * @param conf
     * @return
     */
    public MailHelper initMailHelper(Configuration conf) {
        JavaMailSenderImpl mailSender = new JavaMailSenderImpl();

        // 设置邮件服务主机
        mailSender.setHost(conf.get(JobConfiguration.MAIL_HOST));
        mailSender.setPort(conf.getInt(JobConfiguration.MAIL_PORT, 25));
        // 发送者邮箱的用户名
        mailSender.setUsername(conf.get(JobConfiguration.MAIL_USERNAME));
        // 发送者邮箱的密码
        mailSender.setPassword(conf.get(JobConfiguration.MAIL_PASSWORD));

        // 配置文件，用于实例化java.mail.session
        Properties prop = System.getProperties();

        // 登录SMTP服务器,需要获得授权
        prop.put("mail.smtp.auth", conf.getString(JobConfiguration.MAIL_SMTP_AUTH, "true"));

        prop.put("mail.transport.protocol", conf.getString(JobConfiguration.MAIL_TRANSPORT_PROTOCOL, "smtp"));
        prop.put("mail.smtp.starttls.enable", conf.getString(JobConfiguration.MAIL_SMTP_STARTTLS_ENABLE, "false"));
        prop.put("mail.smtp.timeout", conf.getInt(JobConfiguration.MAIL_SMTP_TIMEOUT, 5000));
        prop.put("mail.debug", conf.getString(JobConfiguration.MAIL_DEBUG, "false"));

        mailSender.setJavaMailProperties(prop);

        return new MailHelper(mailSender, conf);
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

    public Dispatcher getDispatcher() {
        return dispatcher;
    }

    public Configuration getConf() {
        return conf;
    }

    public ScheduleDataManager getScheduleDataManager() {
        return scheduleDataManager;
    }

    public void setScheduleDataManager(ScheduleDataManager scheduleDataManager) {
        this.scheduleDataManager = scheduleDataManager;
    }

    public ConsoleDataManager getConsoleDataManager() {
        return consoleDataManager;
    }

    public void setConsoleDataManager(ConsoleDataManager consoleDataManager) {
        this.consoleDataManager = consoleDataManager;
    }

    public void setSmsSender(SmsSender smsSender) {
        this.smsSender = smsSender;
    }

    public Dispatcher getDataDispatcher() {
        return dataDispatcher;
    }

    public String getUuid() {
        return this.uuid;
    }
}
